package com.atguigu.sparkcore.rdd.tranformdoublevalue

import com.atguigu.sparkcore.util.MySparkContextUtil
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * rdd转换-Intersection
  * 计算两个集合的交集
  * 计算完结果后，会对结果进行重新分区的操作  cogroup(other, defaultPartitioner(self, other))
  */
object Intersection {

  def main(args: Array[String]): Unit = {

    // 获取sparkContext
    val sc = MySparkContextUtil.get(args)

    // 创建rdd
    val rdd: RDD[Int] = sc.makeRDD((1 to 10), 2)
    val rdd2: RDD[Int] = sc.makeRDD(Array(2, 3, 5), 2)

//    // 转换rdd
//    val result = rdd.intersection(rdd2)
//    // 输出
//    println(result.getNumPartitions)
//    result.collect().foreach(println)
//    result.mapPartitionsWithIndex((index, item)=>item.map((index,_))).foreach(println)

    // 实现原理
    val rdd11 = rdd.map(v => (v, null))
    rdd11.foreach(println)

    val rdd22 = rdd2.map(v => (v, null))
    rdd22.foreach(println)

    val rdd12 = rdd11.cogroup(rdd22)
    rdd12.foreach(println)

    val result = rdd12.filter { case (_, (leftGroup, rightGroup)) => leftGroup.nonEmpty && rightGroup.nonEmpty }.keys
    result.foreach(println)
    
    // 关闭资源
    MySparkContextUtil.close(sc)
  }
}
